/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.store;

import io.opentelemetry.api.common.AttributesBuilder;
import io.opentelemetry.api.metrics.Meter;
import io.opentelemetry.sdk.metrics.InstrumentSelector;
import io.opentelemetry.sdk.metrics.ViewBuilder;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;

import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.SystemClock;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageExtBatch;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.ha.HAService;
import org.apache.rocketmq.store.hook.PutMessageHook;
import org.apache.rocketmq.store.hook.SendMessageBackHook;
import org.apache.rocketmq.store.logfile.MappedFile;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.ConsumeQueueStore;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.apache.rocketmq.store.timer.TimerMessageStore;
import org.apache.rocketmq.store.util.PerfCounter;

/**
 * 该类定义了抽象的接口，允许第三方厂商使用自定义的存储实现。
 * 该接口的实现：
 * - {@class org.apache.rocketmq.store.plugin.AbstractPluginMessageStore}
 * - {@class org.apache.rocketmq.store.DefaultMessageStore}
 * - {@class org.apache.rocketmq.tieredstore.TieredMessageStore}
 */
public interface MessageStore {

    /**
     * 加载已经存储的消息。
     *
     * @return true表示成功，false表示失败。
     */
    boolean load();

    /**
     * 启动该消息存储。
     * 两个类的实现：
     * - {@class org.apache.rocketmq.store.DefaultMessageStore}
     * - {@class org.apache.rocketmq.store.plugin.AbstractPluginMessageStore}
     *
     * @throws Exception 如果有错误则抛异常。
     */
    void start() throws Exception;

    /**
     * 关闭当前消息存储。
     * 三个类的实现：
     * - {@class org.apache.rocketmq.store.plugin.AbstractPluginMessageStore}
     * - {@class org.apache.rocketmq.store.DefaultMessageStore}
     * - {@class org.apache.rocketmq.tieredstore.TieredMessageStore}
     */
    void shutdown();

    /**
     * 销毁当前消息存储。通常，在调用后，所有持久文件都应该被删除。
     * 三个类的实现：
     * - {@class org.apache.rocketmq.store.plugin.AbstractPluginMessageStore}
     * - {@class org.apache.rocketmq.store.DefaultMessageStore}
     * - {@class org.apache.rocketmq.tieredstore.TieredMessageStore}
     */
    void destroy();

    /**
     * 以异步的方式将消息保存到消息存储，处理器无需等待消息存储的完成就可以接着处理下一个请求，同时以异步的方式通知客户端。
     *
     * @param msg 需要保存的MessageInstance实例
     * @return 返回存储消息结果的CompletableFuture。
     */
    default CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
        return CompletableFuture.completedFuture(putMessage(msg));
    }

    /**
     * 以异步方式批量存储消息。
     *
     * @param messageExtBatch 消息批。
     * @return 返回消息批存储结果的CompletableFuture。
     */
    default CompletableFuture<PutMessageResult> asyncPutMessages(final MessageExtBatch messageExtBatch) {
        return CompletableFuture.completedFuture(putMessages(messageExtBatch));
    }

    /**
     * 以同步方式将消息保存到消息存储。
     *
     * @param msg 需要保存的Message实例。
     * @return 返回存储消息结果。
     */
    PutMessageResult putMessage(final MessageExtBrokerInner msg);

    /**
     * 同步方式保存消息批。
     *
     * @param messageExtBatch 消息批。
     * @return 返回保存消息批的结果。
     */
    PutMessageResult putMessages(final MessageExtBatch messageExtBatch);

    /**
     * 从指定主题的指定消息队列，从起始偏移量开始查询最多maxMsgNums条消息。
     * 查询到的消息再通过指定消息过滤器的过滤后，返回。
     *
     * @param group         发起该查询请求的消费组名称。
     * @param topic         要查询的主题名称。
     * @param queueId       要查询的消息队列ID。
     * @param offset        查询消息的偏移量起始值。
     * @param maxMsgNums    查询消息的最大条数。
     * @param messageFilter 对查询到的消息进行过滤的消息过滤器。
     * @return 返回查询到的匹配的消息。
     */
    GetMessageResult getMessage(final String group, final String topic, final int queueId,
                                final long offset, final int maxMsgNums, final MessageFilter messageFilter);

    /**
     * 以异步方式查询消息。
     *
     * @param group         发起本查询的消费组名称。
     * @param topic         主题名称。
     * @param queueId       队列ID。
     * @param offset        查询消息的起始偏移量。
     * @param maxMsgNums    查询消息的最大条数。
     * @param messageFilter 消息过滤器。
     * @return 匹配的消息（一般是多条）
     * @see #getMessage(String, String, int, long, int, MessageFilter) getMessage
     */
    CompletableFuture<GetMessageResult> getMessageAsync(final String group, final String topic, final int queueId,
                                                        final long offset, final int maxMsgNums, final MessageFilter messageFilter);

    /**
     * 从指定主题指定队列的指定偏移量开始查询最多maxMsgNums条消息。
     * 查询到的消息再经过消息过滤器的处理，返回结果。
     * 该方法添加了查询到的消息的最大字节数限制。
     *
     * @param group           发起本次查询的消费组名称。
     * @param topic           主题名称。
     * @param queueId         队列ID。
     * @param offset          起始逻辑偏移量。
     * @param maxMsgNums      要查询消息的最大条数。
     * @param maxTotalMsgSize 要查询消息的最大字节数。
     * @param messageFilter   消息过滤器。
     * @return 返回匹配的消息（一般多条）。
     */
    GetMessageResult getMessage(final String group, final String topic, final int queueId,
                                final long offset, final int maxMsgNums, final int maxTotalMsgSize, final MessageFilter messageFilter);

    /**
     * 以异步方式从指定主题指定消息队列指定的起始偏移量开始查询消息。
     * 查询到的消息再经过消息过滤器的处理后，返回。
     * 该方法添加了对查询到的消息的最大字节数的限制。
     *
     * @param group           发起查询请求的消费组名称。
     * @param topic           主题名称。
     * @param queueId         消息队列ID。
     * @param offset          起始逻辑偏移量。
     * @param maxMsgNums      查询的消息的最大数量。
     * @param maxTotalMsgSize 查询的消息的最大字节数。
     * @param messageFilter   消息过滤器。
     * @return 返回查询到的匹配消息（一般是多条）。
     * @see #getMessage(String, String, int, long, int, int, MessageFilter) getMessage
     */
    CompletableFuture<GetMessageResult> getMessageAsync(final String group, final String topic, final int queueId,
                                                        final long offset, final int maxMsgNums, final int maxTotalMsgSize, final MessageFilter messageFilter);

    /**
     * 获取主题队列的最大偏移量。
     *
     * @param topic   主题名称。
     * @param queueId 消息队列ID。
     * @return 返回当前消息队列的最大偏移量。
     */
    long getMaxOffsetInQueue(final String topic, final int queueId);

    /**
     * 查询主题队列的最大偏移量。
     *
     * @param topic     主题名称。
     * @param queueId   消息队列ID。
     * @param committed true表示返回ConsumeQueue的最大偏移量，false表示返回提交日志的最大偏移量。
     * @return 返回当前的最大偏移量。
     */
    long getMaxOffsetInQueue(final String topic, final int queueId, final boolean committed);

    /**
     * 查询主题队列的最小偏移量。
     *
     * @param topic   主题名称。
     * @param queueId 消息队列ID。
     * @return 返回当前偏移量的最小值。
     */
    long getMinOffsetInQueue(final String topic, final int queueId);

    TimerMessageStore getTimerMessageStore();

    void setTimerMessageStore(TimerMessageStore timerMessageStore);

    /**
     * 查询提交日志的消息偏移量，也称为物理偏移量。
     *
     * @param topic              主题名称
     * @param queueId            队列ID
     * @param consumeQueueOffset 消费队列偏移量
     * @return 物理偏移量。
     */
    long getCommitLogOffsetInQueue(final String topic, final int queueId, final long consumeQueueOffset);

    /**
     * 查询指定存储时间戳的消息的物理偏移量。
     *
     * @param topic     主题名称
     * @param queueId   队列ID
     * @param timestamp 要查询的时间戳。
     * @return 返回匹配的物理偏移量。
     */
    long getOffsetInQueueByTime(final String topic, final int queueId, final long timestamp);

    /**
     * 根据指定的存储时间戳和边界类型（LOWER、UPPER）查询消息的物理偏移量。
     *
     * @param topic        主题名称。
     * @param queueId      消息队列ID。
     * @param timestamp    时间戳
     * @param boundaryType 边界类型，包括：LOWER、UPPER
     * @return 返回匹配的物理偏移量。
     */
    long getOffsetInQueueByTime(final String topic, final int queueId, final long timestamp, final BoundaryType boundaryType);

    /**
     * 根据提交日志的偏移量查询消息。
     *
     * @param commitLogOffset 物理偏移量
     * @return 返回指定物理偏移量对应的消息。
     */
    MessageExt lookMessageByOffset(final long commitLogOffset);

    /**
     * 根据偏移量和消息大小，查询对应的消息。
     *
     * @param commitLogOffset 物理偏移量
     * @param size            消息大小
     * @return 指定物理偏移量位置的消息实例。
     */
    MessageExt lookMessageByOffset(long commitLogOffset, int size);

    /**
     * 查询提交日志指定偏移量处的消息。
     *
     * @param commitLogOffset 提交日志偏移量
     * @return 消息的封装对象。
     */
    SelectMappedBufferResult selectOneMessageByOffset(final long commitLogOffset);

    /**
     * 从提交日志指定偏移量处查询一条消息。
     *
     * @param commitLogOffset 提交日志偏移量。
     * @param msgSize         消息字节数。
     * @return 消息的封装对象。
     */
    SelectMappedBufferResult selectOneMessageByOffset(final long commitLogOffset, final int msgSize);

    /**
     * 获取当前store的运行信息。
     *
     * @return 返回消息存储的运行状态信息。
     */
    String getRunningDataInfo();

    long getTimingMessageCount(String topic);

    /**
     * 查询消息存储的运行时信息，一般应包含各种各样的统计信息。
     *
     * @return 以键值对的形式返回消息存储的运行时信息。
     */
    HashMap<String, String> getRuntimeInfo();

    /**
     * 查询高可用运行时信息。
     * @return 高可用运行时信息。
     */
    HARuntimeInfo getHARuntimeInfo();

    /**
     * 查询提交日志的最大偏移量。
     * @return 返回提交日志的最大偏移量
     */
    long getMaxPhyOffset();

    /**
     * 查询提交日志最小偏移量
     * @return 返回提交日志的最小偏移量。
     */
    long getMinPhyOffset();

    /**
     * 查询指定消息队列中最早的消息的存储时间戳。
     *
     * @param topic   消息队列所属的主题名称。
     * @param queueId 消息队列的ID。
     * @return 返回队列中最早消息的存储时间戳。
     */
    long getEarliestMessageTime(final String topic, final int queueId);

    /**
     * 获取日志存储中最早消息的存储时间戳
     * @return 返回消息存储中最早消息的存储时间戳。
     */
    long getEarliestMessageTime();

    /**
     * Asynchronous get the store time of the earliest message in this store.
     *
     * @return timestamp of the earliest message in this store.
     * @see #getEarliestMessageTime() getEarliestMessageTime
     */
    CompletableFuture<Long> getEarliestMessageTimeAsync(final String topic, final int queueId);

    /**
     * 查询指定消息的存储时间戳。
     * @param topic              消息主题名称。
     * @param queueId            消息队列ID。
     * @param consumeQueueOffset 消费队列偏移量。
     * @return 消息的存储时间戳。
     */
    long getMessageStoreTimeStamp(final String topic, final int queueId, final long consumeQueueOffset);

    /**
     * 以异步方式查询指定消息的存储时间。
     * @param topic              消息主题名称。
     * @param queueId            消息队列ID。
     * @param consumeQueueOffset 消费队列偏移量。
     * @return 消息的存储时间戳。
     * @see #getMessageStoreTimeStamp(String, int, long) getMessageStoreTimeStamp
     */
    CompletableFuture<Long> getMessageStoreTimeStampAsync(final String topic, final int queueId,
                                                          final long consumeQueueOffset);

    /**
     * 查询指定消息队列中的消息总量。
     * @param topic   主题名称。
     * @param queueId 消息队列ID。
     * @return 返回该消息队列中的消息总量。
     */
    long getMessageTotalInQueue(final String topic, final int queueId);

    /**
     * Get the raw commit log data starting from the given offset, which should used for replication purpose.
     *
     * @param offset starting offset.
     * @return commit log data.
     */
    SelectMappedBufferResult getCommitLogData(final long offset);

    /**
     * Get the raw commit log data starting from the given offset, across multiple mapped files.
     *
     * @param offset starting offset.
     * @param size   size of data to get
     * @return commit log data.
     */
    List<SelectMappedBufferResult> getBulkCommitLogData(final long offset, final int size);

    /**
     * 向消息存储追加数据。
     * @param startOffset 其实偏移量。
     * @param data        要追加的数据。
     * @param dataStart   要追加的数据在字节数组data中的起始位置。
     * @param dataLength  要追加的数据在字节数组data中的长度。
     * @return true表示成功，false表示失败。
     */
    boolean appendToCommitLog(final long startOffset, final byte[] data, int dataStart, int dataLength);

    /**
     * 手动执行文件删除。
     */
    void executeDeleteFilesManually();

    /**
     * 根据给定的key查询消息。
     *
     * @param topic  消息所属的主题名称。
     * @param key    消息的key。
     * @param maxNum 最大返回的消息数量。
     * @param begin  起始时间戳。
     * @param end    结束时间戳。
     */
    QueryMessageResult queryMessage(final String topic, final String key, final int maxNum, final long begin,
                                    final long end);

    /**
     * Asynchronous query messages by given key.
     *
     * @param topic  topic of the message.
     * @param key    message key.
     * @param maxNum maximum number of the messages possible.
     * @param begin  begin timestamp.
     * @param end    end timestamp.
     * @see #queryMessage(String, String, int, long, long) queryMessage
     */
    CompletableFuture<QueryMessageResult> queryMessageAsync(final String topic, final String key, final int maxNum,
                                                            final long begin, final long end);

    /**
     * Update HA master address.
     *
     * @param newAddr new address.
     */
    void updateHaMasterAddress(final String newAddr);

    /**
     * Update master address.
     *
     * @param newAddr new address.
     */
    void updateMasterAddress(final String newAddr);

    /**
     * Return how much the slave falls behind.
     *
     * @return number of bytes that slave falls behind.
     */
    long slaveFallBehindMuch();

    /**
     * 返回消息存储的当前时间戳。
     *
     * @return 返回从1970-01-01开始的毫秒值。
     */
    long now();

    /**
     * 删除主题的消费队列文件以及不再使用的指标。
     * 该接口允许用户删除系统主题。
     *
     * @param deleteTopics 不再使用的主题名称集合。
     * @return 删除的topic数量。
     */
    int deleteTopics(final Set<String> deleteTopics);

    /**
     * Clean unused topics which not in retain topic name set.
     *
     * @param retainTopics all valid topics.
     * @return number of the topics deleted.
     */
    int cleanUnusedTopic(final Set<String> retainTopics);

    /**
     * Clean expired consume queues.
     */
    void cleanExpiredConsumerQueue();

    /**
     * Check if the given message has been swapped out of the memory.
     *
     * @param topic         topic.
     * @param queueId       queue ID.
     * @param consumeOffset consume queue offset.
     * @return true if the message is no longer in memory; false otherwise.
     * @deprecated As of RIP-57, replaced by {@link #checkInMemByConsumeOffset(String, int, long, int)}, see <a href="https://github.com/apache/rocketmq/issues/5837">this issue</a> for more details
     */
    @Deprecated
    boolean checkInDiskByConsumeOffset(final String topic, final int queueId, long consumeOffset);

    /**
     * Check if the given message is in the page cache.
     *
     * @param topic         topic.
     * @param queueId       queue ID.
     * @param consumeOffset consume queue offset.
     * @return true if the message is in page cache; false otherwise.
     */
    boolean checkInMemByConsumeOffset(final String topic, final int queueId, long consumeOffset, int batchSize);

    /**
     * Check if the given message is in store.
     *
     * @param topic         topic.
     * @param queueId       queue ID.
     * @param consumeOffset consume queue offset.
     * @return true if the message is in store; false otherwise.
     */
    boolean checkInStoreByConsumeOffset(final String topic, final int queueId, long consumeOffset);

    /**
     * Get number of the bytes that have been stored in commit log and not yet dispatched to consume queue.
     *
     * @return number of the bytes to dispatch.
     */
    long dispatchBehindBytes();

    /**
     * Flush the message store to persist all data.
     *
     * @return maximum offset flushed to persistent storage device.
     */
    long flush();

    /**
     * Get the current flushed offset.
     *
     * @return flushed offset
     */
    long getFlushedWhere();

    /**
     * Reset written offset.
     *
     * @param phyOffset new offset.
     * @return true if success; false otherwise.
     */
    boolean resetWriteOffset(long phyOffset);

    /**
     * Get confirm offset.
     *
     * @return confirm offset.
     */
    long getConfirmOffset();

    /**
     * Set confirm offset.
     *
     * @param phyOffset confirm offset to set.
     */
    void setConfirmOffset(long phyOffset);

    /**
     * Check if the operation system page cache is busy or not.
     *
     * @return true if the OS page cache is busy; false otherwise.
     */
    boolean isOSPageCacheBusy();

    /**
     * Get lock time in milliseconds of the store by far.
     *
     * @return lock time in milliseconds.
     */
    long lockTimeMills();

    /**
     * Check if the transient store pool is deficient.
     *
     * @return true if the transient store pool is running out; false otherwise.
     */
    boolean isTransientStorePoolDeficient();

    /**
     * 获取分发器列表。
     *
     * @return 分发器列表。
     */
    LinkedList<CommitLogDispatcher> getDispatcherList();

    /**
     * 添加分发器。
     *
     * @param dispatcher 要添加的提交日志分发器。
     */
    void addDispatcher(CommitLogDispatcher dispatcher);

    /**
     * 查询主题/队列的消费队列。如果没找到，则返回null。
     *
     * @param topic   主题名称。
     * @param queueId 消息队列ID。
     * @return 返回匹配的消息队列。
     */
    ConsumeQueueInterface getConsumeQueue(String topic, int queueId);

    /**
     * Get consume queue of the topic/queue. If consume queue not exist, will create one then return it.
     *
     * @param topic   Topic.
     * @param queueId Queue ID.
     * @return Consume queue.
     */
    ConsumeQueueInterface findConsumeQueue(String topic, int queueId);

    /**
     * Get BrokerStatsManager of the messageStore.
     *
     * @return BrokerStatsManager.
     */
    BrokerStatsManager getBrokerStatsManager();

    /**
     * Will be triggered when a new message is appended to commit log.
     *
     * @param msg           the msg that is appended to commit log
     * @param result        append message result
     * @param commitLogFile commit log file
     */
    void onCommitLogAppend(MessageExtBrokerInner msg, AppendMessageResult result, MappedFile commitLogFile);

    /**
     * Will be triggered when a new dispatch request is sent to message store.
     *
     * @param dispatchRequest dispatch request
     * @param doDispatch      do dispatch if true
     * @param commitLogFile   commit log file
     * @param isRecover       is from recover process
     * @param isFileEnd       if the dispatch request represents 'file end'
     */
    void onCommitLogDispatch(DispatchRequest dispatchRequest, boolean doDispatch, MappedFile commitLogFile,
                             boolean isRecover, boolean isFileEnd);

    /**
     * Get the message store config
     *
     * @return the message store config
     */
    MessageStoreConfig getMessageStoreConfig();

    /**
     * Get the statistics service
     *
     * @return the statistics service
     */
    StoreStatsService getStoreStatsService();

    /**
     * Get the store checkpoint component
     *
     * @return the checkpoint component
     */
    StoreCheckpoint getStoreCheckpoint();

    /**
     * 获取系统时钟。
     *
     * @return the system clock
     */
    SystemClock getSystemClock();

    /**
     * 获取CommitLog
     *
     * @return the commit log
     */
    CommitLog getCommitLog();

    /**
     * Get running flags
     *
     * @return running flags
     */
    RunningFlags getRunningFlags();

    /**
     * Get the transient store pool
     *
     * @return the transient store pool
     */
    TransientStorePool getTransientStorePool();

    /**
     * Get the HA service
     *
     * @return the HA service
     */
    HAService getHaService();

    /**
     * Get the allocate-mappedFile service
     *
     * @return the allocate-mappedFile service
     */
    AllocateMappedFileService getAllocateMappedFileService();

    /**
     * Truncate dirty logic files
     *
     * @param phyOffset physical offset
     */
    void truncateDirtyLogicFiles(long phyOffset);

    /**
     * Destroy logics files
     */
    void destroyLogics();

    /**
     * Unlock mappedFile
     *
     * @param unlockMappedFile the file that needs to be unlocked
     */
    void unlockMappedFile(MappedFile unlockMappedFile);

    /**
     * Get the perf counter component
     *
     * @return the perf counter component
     */
    PerfCounter.Ticks getPerfCounter();

    /**
     * Get the queue store
     *
     * @return the queue store
     */
    ConsumeQueueStore getQueueStore();

    /**
     * If 'sync disk flush' is configured in this message store
     *
     * @return yes if true, no if false
     */
    boolean isSyncDiskFlush();

    /**
     * If this message store is sync master role
     *
     * @return yes if true, no if false
     */
    boolean isSyncMaster();

    /**
     * Assign a message to queue offset. If there is a race condition, you need to lock/unlock this method
     * yourself.
     *
     * @param msg message
     */
    void assignOffset(MessageExtBrokerInner msg);

    /**
     * Increase queue offset in memory table. If there is a race condition, you need to lock/unlock this method
     *
     * @param msg        message
     * @param messageNum message num
     */
    void increaseOffset(MessageExtBrokerInner msg, short messageNum);

    /**
     * Get master broker message store in process in broker container
     *
     * @return
     */
    MessageStore getMasterStoreInProcess();

    /**
     * Set master broker message store in process
     *
     * @param masterStoreInProcess
     */
    void setMasterStoreInProcess(MessageStore masterStoreInProcess);

    /**
     * Use FileChannel to get data
     *
     * @param offset
     * @param size
     * @param byteBuffer
     * @return
     */
    boolean getData(long offset, int size, ByteBuffer byteBuffer);

    /**
     * Set the number of alive replicas in group.
     *
     * @param aliveReplicaNums number of alive replicas
     */
    void setAliveReplicaNumInGroup(int aliveReplicaNums);

    /**
     * Get the number of alive replicas in group.
     *
     * @return number of alive replicas
     */
    int getAliveReplicaNumInGroup();

    /**
     * Wake up AutoRecoverHAClient to start HA connection.
     */
    void wakeupHAClient();

    /**
     * Get master flushed offset.
     *
     * @return master flushed offset
     */
    long getMasterFlushedOffset();

    /**
     * Get broker init max offset.
     *
     * @return broker max offset in startup
     */
    long getBrokerInitMaxOffset();

    /**
     * Set master flushed offset.
     *
     * @param masterFlushedOffset master flushed offset
     */
    void setMasterFlushedOffset(long masterFlushedOffset);

    /**
     * Set broker init max offset.
     *
     * @param brokerInitMaxOffset broker init max offset
     */
    void setBrokerInitMaxOffset(long brokerInitMaxOffset);

    /**
     * Calculate the checksum of a certain range of data.
     *
     * @param from begin offset
     * @param to   end offset
     * @return checksum
     */
    byte[] calcDeltaChecksum(long from, long to);

    /**
     * Truncate commitLog and consume queue to certain offset.
     *
     * @param offsetToTruncate offset to truncate
     * @return true if truncate succeed, false otherwise
     */
    boolean truncateFiles(long offsetToTruncate);

    /**
     * Check if the offset is align with one message.
     *
     * @param offset offset to check
     * @return true if align, false otherwise
     */
    boolean isOffsetAligned(long offset);

    /**
     * Get put message hook list
     *
     * @return List of PutMessageHook
     */
    List<PutMessageHook> getPutMessageHookList();

    /**
     * Set send message back hook
     *
     * @param sendMessageBackHook
     */
    void setSendMessageBackHook(SendMessageBackHook sendMessageBackHook);

    /**
     * Get send message back hook
     *
     * @return SendMessageBackHook
     */
    SendMessageBackHook getSendMessageBackHook();

    //The following interfaces are used for duplication mode

    /**
     * Get last mapped file and return lase file first Offset
     *
     * @return lastMappedFile first Offset
     */
    long getLastFileFromOffset();

    /**
     * Get last mapped file
     *
     * @param startOffset
     * @return true when get the last mapped file, false when get null
     */
    boolean getLastMappedFile(long startOffset);

    /**
     * Set physical offset
     *
     * @param phyOffset
     */
    void setPhysicalOffset(long phyOffset);

    /**
     * Return whether mapped file is empty
     *
     * @return whether mapped file is empty
     */
    boolean isMappedFilesEmpty();

    /**
     * Get state machine version
     *
     * @return state machine version
     */
    long getStateMachineVersion();

    /**
     * Check message and return size
     *
     * @param byteBuffer
     * @param checkCRC
     * @param checkDupInfo
     * @param readBody
     * @return DispatchRequest
     */
    DispatchRequest checkMessageAndReturnSize(final ByteBuffer byteBuffer, final boolean checkCRC,
                                              final boolean checkDupInfo, final boolean readBody);

    /**
     * Get remain transientStoreBuffer numbers
     *
     * @return remain transientStoreBuffer numbers
     */
    int remainTransientStoreBufferNumbs();

    /**
     * Get remain how many data to commit
     *
     * @return remain how many data to commit
     */
    long remainHowManyDataToCommit();

    /**
     * Get remain how many data to flush
     *
     * @return remain how many data to flush
     */
    long remainHowManyDataToFlush();

    /**
     * Get whether message store is shutdown
     *
     * @return whether shutdown
     */
    boolean isShutdown();

    /**
     * Estimate number of messages, within [from, to], which match given filter
     *
     * @param topic   Topic name
     * @param queueId Queue ID
     * @param from    Lower boundary of the range, inclusive.
     * @param to      Upper boundary of the range, inclusive.
     * @param filter  The message filter.
     * @return Estimate number of messages matching given filter.
     */
    long estimateMessageCount(String topic, int queueId, long from, long to, MessageFilter filter);

    /**
     * 获取消息存储的指标数据。
     *
     * @return List of metrics selector and view pair
     */
    List<Pair<InstrumentSelector, ViewBuilder>> getMetricsView();

    /**
     * 初始化消息存储指标。
     *
     * @param meter                     opentelemetry meter
     * @param attributesBuilderSupplier metrics attributes builder
     */
    void initMetrics(Meter meter, Supplier<AttributesBuilder> attributesBuilderSupplier);
}
